/**
 *******************************************************************************
 * \brief   CAS EBX Project Work "Kaffeemaschine"
 *
 * \file    channel.c
 * \version 1.0
 * \date    30.06.2010
 * \author  Baehler Thomas, Liechti Beat, Sauter Marc
 * \author  Vuksic Valentina, Zbinden Rene
 *
 * \remark  Last Modifications:
 * \remark  V1.0, 30.06.2010   Initial release
 * \remark  V1.1, 01.07.2010   First working implementation of channel with fifo
 * \remark  V1.2, 01.07.2010   Added R & W semaphore, blocking write (bug)
 * \remark  V1.3, 07.07.2010   Using message queue instead of fifo
 * \remark  V1.4  12.07.2010   Updating documentation for doxygen
 *******************************************************************************
 **/

/*
 *******************************************************************************
 * Imports
 *******************************************************************************
 */

#include <errno.h>
#include <fcntl.h>		/* For O_* constants */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>	/* For mode constants */
#include <unistd.h>   

#include "channel.h"

/*
 *******************************************************************************
 * Definitions
 *******************************************************************************
 */

#define ROOT_PATH_MQ		"/channelMq_"
#define SEM_R 				"/channelMqR_"
#define SEM_W 				"/channelMqW_"
#define CHANNEL_MAX_ITEMS	10

/*
 *******************************************************************************
 * Local Function Declaration
 *******************************************************************************
 */

/**
 *******************************************************************************
 * mallocChannel
 *
 * \brief allocates required memory a channel ADT struct
 *
 * \param name The name of the channel
 * \param itemsize The size of the data item
 *
 * \return tChannel* Pointer to channel ADT structure, NULL if error
 *
 *******************************************************************************
 **/

tChannel *mallocChannel(char *name, int itemsize);

/**
 *******************************************************************************
 * freeChannel
 *
 * \brief releases memory of a channel ADT struct
 *
 * \param channel Pointer to channel ADT structure
 *
 *******************************************************************************
 **/

void freeChannel(tChannel *channel);

/*
 *******************************************************************************
 * Global Function Definition
 *******************************************************************************
 */

/**
 *******************************************************************************
 * createChannel
 *
 * \brief creates a channel
 *
 * \param name The name of the channel
 * \param itemsize The size of the data item
 *
 * \return int Zero if channel has been created successfully, -1 otherwise
 *
 *******************************************************************************
 **/

int createChannel(char *name, int itemsize)
{
	int ret = -1;
	/* build path from name */
	int pathlen = strlen(ROOT_PATH_MQ) + strlen(name) + 1;
	char *path = alloca(pathlen);
	sprintf(path, "%s%s", ROOT_PATH_MQ, name);
	/* remove legacy queue */
	mq_unlink(path);
	/* create message queue */
	mqd_t queue;
	struct mq_attr attrs;

	memset(&attrs, 0, sizeof (struct mq_attr));
	attrs.mq_msgsize = itemsize;
	attrs.mq_maxmsg = CHANNEL_MAX_ITEMS;

	if ((queue = mq_open(path, O_CREAT | O_RDWR,  S_IRWXU | S_IRWXG, &attrs)) < 0)
	{
		perror("error creating queue");
		printf("failed to create queue %s, retval %d\n", path, queue);
		ret = queue;
	}
	else
	{
		printf("created message queue %s\n", path);
		mq_close(queue);
		/* create read semaphore */
		int semNameLenRead = strlen(SEM_R) + strlen(name) + 1;
		char *semNameRead = alloca(semNameLenRead);
		sprintf(semNameRead, "%s%s", SEM_R, name);
		sem_unlink(semNameRead);
		sem_t *semRead = sem_open(semNameRead, O_CREAT, S_IRUSR | S_IWUSR, 1);
		if (semRead == SEM_FAILED)
		{
			perror("sem_open for semRead failed");
			mq_unlink(path);
		}
		else
		{
			printf("created semaphore %s\n", semNameRead);
			sem_close(semRead);
			/* create write semaphore */
			int semNameLenWrite = strlen(SEM_W) + strlen(name) + 1;
			char *semNameWrite = alloca(semNameLenWrite);
			sprintf(semNameWrite, "%s%s", SEM_W, name);
			sem_unlink(semNameWrite);
			sem_t *semWrite = sem_open(semNameWrite, O_CREAT, S_IRWXU | S_IRWXG, 1);
			if (semWrite == SEM_FAILED)
			{
				perror("sem_open for semWrite failed");
				sem_unlink(semNameRead);
				mq_unlink(path);
			}
			else
			{
				printf("created semaphore %s\n", semNameWrite);
				sem_close(semWrite);
				ret = 0;
			}
		}
	}
	return ret;
}

/**
 *******************************************************************************
 * deleteChannel
 *
 * \brief deletes a channel
 *
 * \param name The name of the channel
 *
 *******************************************************************************
 **/

void deleteChannel(char *name)
{
	/* delete write semaphore */
	int semNameLenWrite = strlen(SEM_W) + strlen(name) + 1;
	char *semNameWrite = alloca(semNameLenWrite);
	sprintf(semNameWrite, "%s%s", SEM_W, name);
	sem_unlink(semNameWrite);
	printf("unlinked semaphore %s\n", semNameWrite);
	/* delete read semaphore */
	int semNameLenRead = strlen(SEM_R) + strlen(name) + 1;
	char *semNameRead = alloca(semNameLenRead);
	sprintf(semNameRead, "%s%s", SEM_R, name);
	sem_unlink(semNameRead);
	printf("unlinked semaphore %s\n", semNameRead);
	/* delete named pipe */
	int pathlen = strlen(ROOT_PATH_MQ) + strlen(name) + 1;
	char *path = alloca(pathlen);
	sprintf(path, "%s%s", ROOT_PATH_MQ, name);
	mq_unlink(path);
	printf("removed message queue %s\n", path);
}

/**
 *******************************************************************************
 * openChannelRead
 *
 * \brief opens a channel for reading
 *
 * \param name The name of the channel
 * \param itemsize The size of the data item
 *
 * \return tChannel* Pointer to channel ADT structure, NULL if error
 *
 *******************************************************************************
 **/

tChannel *openChannelRead(char *name, int itemsize)
{
	tChannel *channel = mallocChannel(name, itemsize);
	if (channel != NULL)
	{
		channel->queue = mq_open(channel->path, O_RDONLY | O_NONBLOCK);
		if (channel->queue < 0)
		{
			freeChannel(channel);
			channel = NULL;
		}
	}
	return channel;
}

/**
 *******************************************************************************
 * openChannelWrite
 *
 * \brief opens a channel for writing
 *
 * \param name The name of the channel
 * \param itemsize The size of the data item

 * \return tChannel* Pointer to channel ADT structure, NULL if error
 *
 *******************************************************************************
 **/

tChannel *openChannelWrite(char *name, int itemsize)
{
	tChannel *channel = mallocChannel(name, itemsize);
	if (channel != NULL)
	{
		channel->queue = mq_open(channel->path, O_WRONLY);
		if (channel->queue < 0)
		{
			printf("mq_open failed with error code %x", channel->queue);
			freeChannel(channel);
			channel = NULL;
		}
	}
	return channel;
}

/**
 *******************************************************************************
 * closeChannel
 *
 * \brief closes a channel (releases ADT structure resources)
 *
 * \param channel Pointer to channel ADT structure
 *
 *******************************************************************************
 **/

void closeChannel(tChannel *channel)
{
	if (channel->queue > 0)
	{
		mq_close(channel->queue);
		freeChannel(channel);
		channel = NULL;
	}
}

/**
 *******************************************************************************
 * readChannel
 *
 * \brief reads from a channel
 *
 * \param channel Pointer to channel ADT structure
 * \param data Pointer to destination for data
 *
 * \return int Bytes read, error if < 0
 *
 *******************************************************************************
 **/

int readChannel(tChannel *channel, void *data)
{
	int ret = -1;
	if (sem_wait(channel->semRead) < 0)
	{
		perror("error getting semaphore semRead");
	}
	else
	{
		ret = mq_receive(channel->queue, data, channel->itemsize, NULL);
		if (ret == EAGAIN || ret == EWOULDBLOCK || ret == -1)
		{
			ret = 0;
		}
		if (sem_post(channel->semRead) < 0) 
		{
			perror("Error release semaphore semRead");
		}
	}
	return ret;
}

/**
 *******************************************************************************
 * writeChannel
 *
 * \brief writes to a channel
 *
 * \param channel Pointer to channel ADT structure
 * \param data Pointer to the data to write
 *
 * \return int Bytes written, error if < 0
 *
 *******************************************************************************
 **/

int writeChannel(tChannel *channel, void *data)
{
	int ret = -1;
	if (sem_wait(channel->semWrite) < 0)
	{
		perror("error getting semaphore semWrite");
	}
	else
	{
		ret = mq_send(channel->queue, data, channel->itemsize, 1);
		if (ret == 0)
		{
			ret = channel->itemsize;
		}
		if (sem_post(channel->semWrite) < 0) 
		{
			perror("Error release semaphore semWrite");
		}
	}
	return ret;
}

/*
 *******************************************************************************
 * Local Function Definition
 *******************************************************************************
 */

/**
 *******************************************************************************
 * mallocChannel
 *
 * \brief allocates required memory a channel ADT struct
 *
 * \param name The name of the channel
 * \param itemsize The size of the data item
 *
 * \return tChannel* Pointer to channel ADT structure, NULL if error
 *
 *******************************************************************************
 **/

tChannel *mallocChannel(char *name, int itemsize)
{
	tChannel *channel = malloc(sizeof(tChannel));
	if (channel != NULL)
	{
		int fErr = 1;
		memset(channel, 0, sizeof(tChannel));
		channel->name = strdup(name);
		if (channel->name != NULL)
		{
			int pathlen = strlen(ROOT_PATH_MQ) + strlen(name) + 1;
			channel->path = malloc(pathlen);
			if (channel->path != NULL)
			{
				sprintf(channel->path, "%s%s", ROOT_PATH_MQ, name);
				channel->itemsize = itemsize;
				int semNameLenRead = strlen(SEM_R) + strlen(name) + 1;
				char *semNameRead = alloca(semNameLenRead);
				sprintf(semNameRead, "%s%s", SEM_R, name);
				channel->semRead = sem_open(semNameRead, O_RDWR);
				if (channel->semRead != SEM_FAILED)
				{
					int semNameLenWrite = strlen(SEM_W) + strlen(name) + 1;
					char *semNameWrite = alloca(semNameLenWrite);
					sprintf(semNameWrite, "%s%s", SEM_W, name);
					channel->semWrite = sem_open(semNameWrite, O_RDWR);
					if (channel->semWrite != SEM_FAILED)
					{
						fErr = 0;
					}
				}
			}
		}
		if (fErr != 0)
		{
			freeChannel(channel);
			channel = NULL;
		}
	}
	return channel;
}

/**
 *******************************************************************************
 * freeChannel
 *
 * \brief releases memory of a channel ADT struct
 *
 * \param channel Pointer to channel ADT structure
 *
 *******************************************************************************
 **/

void freeChannel(tChannel *channel)
{
	if (channel != NULL)
	{
		if (channel->semWrite != 0 && channel->semWrite != SEM_FAILED)
		{
			sem_close(channel->semWrite);
		}
		if (channel->semRead != 0 && channel->semRead != SEM_FAILED)
		{
			sem_close(channel->semRead);
		}
		if (channel->name != NULL)
		{
			free(channel->name);
		}
		if (channel->path != NULL)
		{
			free(channel->path);
		}
		free(channel);
	}	
}
